package im.vinci.server.common.tmsg;

import im.vinci.server.utils.JsonUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;

import javax.sql.DataSource;
import java.io.Serializable;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.List;

/**
 * Created by tim@vinci on 17/2/9.
 */
public class SingleDBMessageStore implements Serializable{
    private static final Logger log = LoggerFactory.getLogger(SingleDBMessageStore.class);

    public static final String TABLE_NAME = "tx_msg_queue";
    public static final String insertSQL = String.format("INSERT INTO %s (message_id,create_time,content) VALUES(?,now(),?)", TABLE_NAME);
    public static final String errorSQL = String.format("UPDATE %s SET status=?,error=error+1,update_time=now() WHERE message_id=?", TABLE_NAME);
    public static final String finishSQL = String.format("DELETE FROM %s WHERE message_id=?", TABLE_NAME);
    public static final String selectSQL = String.format("SELECT * from %s WHERE id>0 and (status=1 or create_time<FROM_UNIXTIME(?)) limit 50",TABLE_NAME);

    public static final String deleteLock = String.format("DELETE FROM %s WHERE id=-1 and create_time < FROM_UNIXTIME(?)", TABLE_NAME);
    public static final String tryLock = String.format("INSERT IGNORE INTO %s (id,message_id,create_time,content) VALUES(-1,'0',now(),'')", TABLE_NAME);

    private final DataSource dataSource;
    private final JdbcTemplate platform;

    public SingleDBMessageStore(DataSource datasource)
    {
        this.dataSource = datasource;
        this.platform = new JdbcTemplate(datasource);
    }

    public void insertNew(ProducerMessage message)
    {
        this.platform.update(insertSQL, message.getKey(), JsonUtils.encode(message));
    }

    public void finish(ProducerMessage message)
    {
        this.platform.update(finishSQL, message.getKey());
    }

    public void error(ProducerMessage message)
    {
        this.platform.update(errorSQL, 1, message.getKey());
    }

    public List<ProducerMessage> selectTimeoutMsg() {
        return this.platform.query(selectSQL, new RowMapper<ProducerMessage>() {
            @Override
            public ProducerMessage mapRow(ResultSet rs, int rowNum) throws SQLException {
                return JsonUtils.decode(rs.getString("content"),ProducerMessage.class);
            }
        },System.currentTimeMillis()/1000 - 60);
    }

    public boolean tryLock() {
        try {
            this.platform.update(deleteLock, System.currentTimeMillis() / 1000 - 60);
            int i = this.platform.update(tryLock);
            return i>0;
        }catch (Exception e) {
            return false;
        }
    }

    public void deleteLock() {
        try {
            this.platform.update(deleteLock,Integer.MAX_VALUE);
        }catch (Exception e) {
            log.error("删除事务消息锁失败");
        }
    }
}
